package net.wicp.tams.common.others.kafka;

import java.lang.reflect.ParameterizedType;
import java.util.Arrays;
import java.util.Properties;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

public abstract class KafkaConsumerThread<T> {
	private final KafkaConsumer<String, T> consumer;
	private final String topic;
	private final IConsumer<T> doConsumer;
	private ExecutorService executor;

	public KafkaConsumerThread(String groupId, String topic, IConsumer<T> doConsumer) {
		Properties props = KafkaTools.getProps(false);
		props.put("group.id", groupId);
		props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
		props.put("value.deserializer", KafkaTools.getValueProp(getTClass(), false));
		this.consumer = new KafkaConsumer<String, T>(props);
		this.topic = topic;
		this.doConsumer = doConsumer;
		this.consumer.subscribe(Arrays.asList(this.topic));
	}

	public Class<T> getTClass() {
		Class<T> tClass = (Class<T>) ((ParameterizedType) this.getClass().getGenericSuperclass())
				.getActualTypeArguments()[0];
		return tClass;
	}

	public void start(int threadNumber) {
		executor = new ThreadPoolExecutor(threadNumber, threadNumber, 0L, TimeUnit.MILLISECONDS,
				new ArrayBlockingQueue<Runnable>(1000), new ThreadPoolExecutor.CallerRunsPolicy());
		while (true) {
			ConsumerRecords<String, T> consumerRecords = consumer.poll(100);
			for (final ConsumerRecord<String, T> item : consumerRecords) {
				executor.submit(new Runnable() {
					@Override
					public void run() {
						doConsumer.doWithRecord(item);
					}
				});
			}
		}
	}

}
